package org.mall.modules.mq.order;

/***
 *@author yinxucun
 *@date 2020-01-05 16:54 
 */

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.mall.common.util.json.JacksonUtils;
import org.mall.modules.order.service.OrderService;
import org.mall.modules.pojo.order.dto.ExpireOrderDTO;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * 消息消费监听
 */
@Slf4j
@Component
public class OrderExpireMsgListener implements MessageListenerConcurrently {
    @Autowired
    OrderService orderService;
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext context) {
        if (CollectionUtils.isEmpty(list)){
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
        MessageExt messageExt = list.get(0);
        ExpireOrderDTO expireOrderDTO= (ExpireOrderDTO) JacksonUtils.parseObject(new String(messageExt.getBody()), ExpireOrderDTO.class);
        int reConsume = messageExt.getReconsumeTimes();
        // 消息已经重试了3次，如果不需要再次消费，则返回成功
        if(reConsume ==3){
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
        try {
            orderService.closeExpireOrder(expireOrderDTO.getOrderId());
        } catch (Exception e) {
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        }
        // 消息消费成功
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
}

